
#include "startup_configuration.h"
#include "nosal/log/logger.h"
#include "nosal/log/logging.h"

#include <chrono>
#include <csignal>

//#include <poll.h>
//#include <sys/wait.h>
//#include <sys/prctl.h>

namespace netaos {
namespace nosal {
namespace exec {
namespace internal {

using netaos::nosal::exec::ExecutionState;

namespace
{

typedef std::conditional<std::chrono::high_resolution_clock::is_steady,
    std::chrono::high_resolution_clock,
    std::chrono::steady_clock>::type SteadyClock;

std::int64_t GetTimeSinceEpochInNs()
{
    return static_cast<std::chrono::nanoseconds>(SteadyClock::now().time_since_epoch()).count();
}

netaos::nosal::log::Logger& GetLogger()
{
    static netaos::nosal::log::Logger& logger
        = netaos::nosal::log::CreateLogger("SCPR", "StartupConfiguration", netaos::nosal::exec::internal::kLogLevel);
    return logger;
}

const timespec kPollTimeout{0, 20000000};  // 10 milliseconds

/// @brief Check if scheduling priority is within the implementation limits
///
/// Check if a scheduling priority is within the implementation limits
/// for a particular scheduling policy
///
/// @param absolutePriority the requested scheduling priority
/// @param policy the requested scheduling policy
/// @return true if priority is within limits, false otherwise
///
bool IsPriorityWithinLimits(uint32_t absolutePriority, int policy)
{
    // sched_priority ranges from 0 to 99 on Linux
    if (absolutePriority > 99) {
        return false;
    }

    int priority = static_cast<int>(absolutePriority);

    int minPriority = sched_get_priority_min(policy);
    int maxPriority = sched_get_priority_max(policy);

    if ((minPriority == -1) || (maxPriority == -1)) {
        return false;
    }

    if ((priority < minPriority) || (priority > maxPriority)) {
        return false;
    }

    return true;
}
#if 0
/// @brief Check if core affinity configuration is valid and get the correct cpu_set
///
/// Check if core affinity configuration is valid and get correct cpu_set for
/// core affinity setting
///
/// @param cpu_set core affinity mask for sched_setaffinity()
/// @param core_affinity_shall_run_on core ids the process should run on
/// @param core_affinity_shall_not_run_on core ids the process should not run on
/// @return true if cpu_set is available, false otherwise
///
bool GetCpuSetFromConfig(cpu_set_t& cpu_set,
    const std::vector<std::uint32_t>& core_affinity_shall_run_on,
    const std::vector<std::uint32_t>& core_affinity_shall_not_run_on)
{
    // ShallRunOn and ShallNotRunOn config not exist
    if (core_affinity_shall_run_on.empty() && core_affinity_shall_not_run_on.empty()) {
        return false;
    }
    // ShallRunOn and ShallNotRunOn config can not exist at the same time
    if ((!core_affinity_shall_run_on.empty()) && (!core_affinity_shall_not_run_on.empty())) {
        GetLogger().LogError() << "ShallRunOn and ShallNotRunOn config can not exist at the same time.";
        return false;
    }
    if (!core_affinity_shall_run_on.empty()) {
        // core_affinity_shall_run_on has data
        CPU_ZERO(&cpu_set);
        // Set cpu base on core_affinity_shall_run_on
        for (std::uint32_t cpu_id : core_affinity_shall_run_on) {
            if (cpu_id < CPU_SETSIZE) {
                CPU_SET(int(cpu_id), &cpu_set);
            } else {
                GetLogger().LogError() << "Invalid cpu id:" << cpu_id << "in ShallRunOn config.";
                return false;
            }
        }
    } else {
        // Otherwise set cpu base on core_affinity_shall_not_run_on
        // Make all cpu available because we do not know which cpu is available
        for (int i = 0; i < CPU_SETSIZE; i++) {
            CPU_SET(i, &cpu_set);
        }
        // Set cpu base on core_affinity_shall_not_run_on
        for (std::uint32_t cpu_id : core_affinity_shall_not_run_on) {
            if (cpu_id < CPU_SETSIZE) {
                CPU_CLR(int(cpu_id), &cpu_set);
            } else {
                GetLogger().LogError() << "Invalid cpu id:" << cpu_id << "in ShallNotRunOn config.";
                return false;
            }
        }
    }
    return true;
}
#endif
}  // namespace

StartupConfiguration::StartupConfiguration(
    const std::string& installDir,
    const std::string& executable_path,
    const std::string& executable_name,
    const config::ExecutionManifest& manifest,
    const config::StartupConfiguration& startupConfig,
    const std::vector<std::string>& machine_specific_environment_variables)
    : prefix_(installDir)
    , config_short_name_(startupConfig.GetConfigShortName())
    , executable_path_(executable_path)
    , executable_name_(executable_name)
    , execution_error_(startupConfig.execution_error_)
    , function_cluster_affiliation_(manifest.GetFunctionClusterAffiliation())
    , manifest_(manifest)
    , number_of_restart_attempts_(manifest.GetNumberOfRestartAttempts())
    , pid_(0)
    , scheduling_policy_(startupConfig.GetSchedulingPolicy())
    , scheduling_priority_(startupConfig.GetSchedulingPriority())
    , core_affinity_shall_run_on_(manifest.GetCoreAffinityShallRunOn())
    , core_affinity_shall_not_run_on_(manifest.GetCoreAffinityShallNotRunOn())
    , reporting_behavior_(manifest.GetReportingBehavior())
    , termination_(TerminationCause::kNormalExit)
    , termination_behavior_(termationBehaviorFromString(startupConfig.termination_behavior_string_))
    , state_(ProcessState::kIdle)
    , enter_timestamp_(0)
    , exit_timestamp_(0)
    , is_platform_app_(false)
    //, execStatePipe_(false)
    //, execStateResponsePipe_(true)
    //, stateClientRequestPipe_(false)
    //, stateClientResponsePipe_(true)
    //, execStatePipeOutput_(execStatePipe_.GetReadEndFd())
    //, execStatePipeInput_(execStateResponsePipe_.GetWriteEndFd())
    //, stateClientRequestPipeOutput_(stateClientRequestPipe_.GetReadEndFd())
    //, stateClientResponsePipeInput_(stateClientResponsePipe_.GetWriteEndFd())
{
    is_platform_app_ = manifest.IsPlatformApplication();

    enter_exit_timeout_ = startupConfig.GetEnterExitTimeout();

    arguments_.AddArgument(executable_path_);
    for (const std::string& argument : startupConfig.GetArguments()) {
        arguments_.AddArgument(argument);
    }

    environment_ = machine_specific_environment_variables;
    std::vector<std::string> environmentVariablesFromExecutionManifest = startupConfig.GetEnvironment();
    environment_.insert(environment_.end(),
        environmentVariablesFromExecutionManifest.begin(),
        environmentVariablesFromExecutionManifest.end());

    // check whether Environment Variables have a value set.
    for (size_t i = 0; i < environment_.size(); i++) {
        size_t pos = environment_[i].find('=');
        if (pos != std::string::npos) {
            GetLogger().LogVerbose() << "StartupConfiguration" << GetName()
                                     << "has Process specific Environment Variable with key and value set.";
            continue;
        }
        GetLogger().LogDebug() << "StartupConfiguration" << GetName()
                               << "has Environment Variable with no value set. Setting empty string as value.";
        environment_[i] = environment_[i] + "= ";
    }

    const std::vector<std::string>& dependencies = startupConfig.GetExecutionDependencies();
    for (const auto& dependency : dependencies) {
        size_t pos = dependency.find('.');
        if (pos == std::string::npos) {
            // If not assume RUNNING state
            dependencies_.insert(std::pair<std::string, ProcessState>(dependency, ProcessState::kRunning));
        } else {
            // Extended dependency format
            const std::string application_name = dependency.substr(0, pos);
            const std::string application_state = dependency.substr(pos + 1, std::string::npos);

            static const std::map<std::string, ProcessState> kAapplicationStateMap{
                {"Idle", ProcessState::kIdle},
                {"Starting", ProcessState::kStarting},
                {"Running", ProcessState::kRunning},
                {"Terminating", ProcessState::kTerminating},
                {"Terminated", ProcessState::kTerminated}};

            if (kAapplicationStateMap.count(application_state) == 1) {
                const ProcessState& process_state = kAapplicationStateMap.at(application_state);
                dependencies_.insert(std::pair<std::string, ProcessState>(application_name, process_state));
            } else {
                GetLogger().LogError() << "StartupConfiguration" << GetName() << "has invalid dependency specified."
                                       << "Default dependency state used instead, check state configuration!";
                dependencies_.insert(
                    std::pair<std::string, ProcessState>(application_name, ProcessState::kRunning));
            }
        }
    }

    // map resource group from Execution Manifest file to internal structure
    const char* resource_group_name = startupConfig.GetResourceGroup().c_str();
    GetLogger().LogDebug() << executable_path_ + ":"
                           << "get resource group" << resource_group_name;
#if 0
    cgroup_ = cgroup_new_cgroup(resource_group_name);
    if (!cgroup_) {
        GetLogger().LogError() << executable_path_ + ":"
                               << "cannot create new cgroup" << resource_group_name;
    }

    int ret = cgroup_get_cgroup(cgroup_);
    if (ret) {
        GetLogger().LogError() << executable_path_ + ":"
                               << "get cgroup failed with" << cgroup_strerror(ret);
    }
#endif
}

StartupConfiguration::StartupConfiguration(StartupConfiguration&& other)
    : prefix_(std::move(other.prefix_))
    , executable_path_(std::move(other.executable_path_))
    , executable_name_(std::move(other.executable_name_))
    , execution_error_(other.execution_error_)
    , manifest_(std::move(other.manifest_))
    , pid_(other.pid_)
    , termination_(other.termination_)
    , state_(other.state_)
    , enter_timestamp_(other.enter_timestamp_)
    , exit_timestamp_(other.exit_timestamp_)
    , is_platform_app_(other.is_platform_app_)
    //, execStatePipe_(std::move(other.execStatePipe_))
    //, execStateResponsePipe_(std::move(other.execStateResponsePipe_))
    //, stateClientRequestPipe_(std::move(other.stateClientRequestPipe_))
    //, stateClientResponsePipe_(std::move(other.stateClientResponsePipe_))
    //, execStatePipeOutput_(execStatePipe_.GetReadEndFd())
    //, execStatePipeInput_(execStateResponsePipe_.GetWriteEndFd())
    //, stateClientRequestPipeOutput_(stateClientRequestPipe_.GetReadEndFd())
    //, stateClientResponsePipeInput_(stateClientResponsePipe_.GetWriteEndFd())
{ }

StartupConfiguration::~StartupConfiguration()
{
    //cgroup_free(&cgroup_);
}

bool StartupConfiguration::Start()
{
    if (pid_ != 0) {
        GetLogger().LogError() << "EM tries to, but cannot start a process that is already running.";
        return false;
    }

    GetLogger().LogInfo() << "Starting executable" << executable_path_;
#if 0
    // These file descriptors will be inherited
    int execClientWriteFd = ::dup(execStatePipe_.GetWriteEndFd());
    int stateClientRequestWriteFd = ::dup(stateClientRequestPipe_.GetWriteEndFd());
    int stateClientResponseReadFd = ::dup(stateClientResponsePipe_.GetReadEndFd());
    int execClientReadFd = ::dup(execStateResponsePipe_.GetReadEndFd());
    if ((-1 == execClientWriteFd) | (-1 == execClientReadFd) | (-1 == stateClientRequestWriteFd)
        | (-1 == stateClientResponseReadFd)) {
        GetLogger().LogError() << "Failed to run child, errno" << errno;
        return false;
    }
#endif
#if 0
    // Prepare cpu_set for core affinity setting.
    cpu_set_t cpu_set;
    bool cpu_set_available = GetCpuSetFromConfig(cpu_set, core_affinity_shall_run_on_, core_affinity_shall_not_run_on_);

    // Prepare scheduling policy and priority
    int schedPolicy = GetSchedulingPolicyFromString(scheduling_policy_);
    if (!IsPriorityWithinLimits(scheduling_priority_, schedPolicy)) {
        GetLogger().LogFatal() << "Error in requested priority for requested policy";
        return false;
    }
    struct sched_param schedParam = {0};
    schedParam.sched_priority = static_cast<int>(scheduling_priority_);
#endif
#if 0
    int pid = ::fork();
    if (pid < 0) {
        GetLogger().LogError() << "Function fork() failed, errno:" << errno;
        return false;
    } else if (pid == 0) {

        // The child process
        GetLogger().LogInfo() << "Child" << GetName() << GetPid() << "started";

        // The child will get a SIGHUP if its parent terminates
        prctl(PR_SET_PDEATHSIG, SIGHUP);

        // Close descriptors which will be used for ExecutionClient and StateClient
        ::close(kExecutionClientWriteFd);
        ::close(kStateClientRequestWriteFd);
        ::close(kStateClientResponseReadFd);
        ::close(kExecutionClientReadFd);

        if ((-1 == ::dup2(execClientWriteFd, kExecutionClientWriteFd))
            | (-1 == ::dup2(execClientReadFd, kExecutionClientReadFd))
            | (-1 == ::dup2(stateClientRequestWriteFd, kStateClientRequestWriteFd))
            | (-1 == ::dup2(stateClientResponseReadFd, kStateClientResponseReadFd)) | (-1 == ::close(execClientWriteFd))
            | (-1 == ::close(execClientReadFd)) | (-1 == ::close(stateClientRequestWriteFd))
            | (-1 == ::close(stateClientResponseReadFd))) {
            GetLogger().LogError() << "Failed to run child, errno" << errno;
            std::abort();
            return false;
        }

        // assign task to defined cgroup
        /// @uptrace{SWS_EM_02102, 5c8d33083da9f51dc92a4615a7fb28549fa899b0}
        /// @uptrace{SWS_EM_02103, 3bbc22939e45afe68713cff972891feb05e81018}
        int ret = cgroup_attach_task(cgroup_);
        if (ret) {
            GetLogger().LogError() << executable_path_ + ":"
                                   << "attach cgroup failed with" << cgroup_strerror(ret);
        }

        // Change scheduling policy and priority
        /// @uptrace{SWS_EM_01014, 43252f0c27e806c34867699bd1fed9a0bbda4b5a}
        /// @uptrace{SWS_EM_01015, cd3006cdcb1797c14052528236c2acaafd178021}
        ret = sched_setscheduler(::getpid(), schedPolicy, &schedParam);
        if (ret) {
            GetLogger().LogError() << GetName() + ": sched_setscheduler failed with error code:" << ret;
        } else {
            GetLogger().LogInfo() << GetName() + ": Set process with schedPolicy:" << scheduling_policy_
                                  << " schedPriority:" << scheduling_priority_;
        }

        // Change core affinity
        /// @uptrace{SWS_EM_02104, 8a1890708965a258b5f4b0816e12f91f43a007eb}
        if (cpu_set_available) {
            ret = sched_setaffinity(0, sizeof(cpu_set), &cpu_set);
            if (ret) {
                GetLogger().LogError() << GetName() << ": sched_setaffinity failed with errno:" << errno;
            }
        }

        const char* childWorkDir = GetPrefix().c_str();

        // Change working directory to application root
        if (::chdir(childWorkDir) != 0) {
            GetLogger().LogFatal() << "Function chdir() failed for dir" << childWorkDir << ',' << "errno: " << errno;
        } else {
            // Set environment variables
            /// @uptrace{SWS_EM_02246, ca680f411ae9785f611b551853be4ac08389f15b}
            /// @uptrace{SWS_EM_02247, 3b8c0cefd3f0858d7234b12b11ae8662aaefe186}
            for (const auto& variable : environment_) {
                ::putenv(const_cast<char*>(variable.c_str()));
            }

            // Redirect terminal output for application to /var/redirected/<executable_name_>
            netaos::nosal::exec::internal::RedirectProcessOutput(executable_name_.c_str());

            const pid_t processPid = ::getpid();

            // TODO: it should be done in the parent process
            // Update pid lookup
            UpdateFindProcessLookup(processPid);

            const char* childPath = executable_path_.c_str();
            char* const* childArgv = const_cast<char* const*>(arguments_.GetArgv());

            // Execute the executable with the specified arguments
            ::execv(childPath, childArgv);

            // When execv() is successful, the current process is replaced by the child.
            // Otherwise, the following code will be reached.
            GetLogger().LogFatal() << "Function execv() failed for executable" << childPath << ',' << "errno:" << errno;

            // TODO: it should be done after wait_pid
            // Update pid lookup
            ResetFindProcessLookup(processPid);
        }

        // Terminate the failed child process
        std::abort();
        return false;
    } else {  // if (pid > 0)
        // The parent process

        ::close(execClientWriteFd);
        ::close(execClientReadFd);
        ::close(stateClientRequestWriteFd);
        ::close(stateClientResponseReadFd);

        /// @uptrace{SWS_EM_02400, 41b89e00faa19e92ef97e16da1d4de6031fef976}
        pid_ = pid;
        enter_timestamp_ = GetTimeSinceEpochInNs();

        if (reporting_behavior_ == config::ExecutionManifest::ExecutionStateReportingBehavior::kDoesNotReport) {
            SetState(ProcessState::kRunning);  ///< @uptrace{SWS_EM_01402, 05207360f588393a90d8c51b4d032db2ec776d55}
        } else /* config::ExecutionManifest::ReportingBehavior::kReports */ {
            SetState(ProcessState::kStarting);  ///< @uptrace{SWS_EM_01003, b1242e7b7614f6168652cfcf2f6c70ba42947793}
        }

        GetLogger().LogInfo() << "Forked child:" << executable_path_ << "with PID" << pid_;
        GetLogger().LogInfo().Flush();

        // The only successfull return
        return true;
    }
#endif
}

void StartupConfiguration::Term()
{
    if (ProcessState::kRunning == state_ || ProcessState::kStarting == state_) {
        SendSignal(SIGTERM);  
        exit_timestamp_ = GetTimeSinceEpochInNs();
    } else {
        GetLogger().LogInfo() << "Ignoring termination request because" << GetName() << "is already being terminated.";
    }
}

void StartupConfiguration::Kill()
{
    if (ProcessState::kTerminated != state_ && ProcessState::kIdle != state_) {
        SendSignal(SIGKILL);
    } else {
        GetLogger().LogInfo() << "Ignoring kill request because process" << GetName() << "is not running.";
    }
}

void StartupConfiguration::SendSignal(int signum)
{
    // Check whether the PID is valid. Otherwise we might kill the entire system
    if (0 == pid_) {
        GetLogger().LogError() << "Ignoring signal" << signum << "to" << executable_name_
                               << "because the process does not exist";

        return;
    }

    GetLogger().LogInfo() << "Sending signal" << signum << "to" << executable_name_ << pid_;

    int result(::kill(pid_, signum));
    if (result != 0) {

        GetLogger().LogError() << "Signal" << signum << "to" << executable_name_
                               << "was ignored because the process does not exist";

        pid_ = 0;
        SetState(ProcessState::kTerminated);
        SetTerminationCause(TerminationCause::kUnknown);
    } else {

        GetLogger().LogInfo() << "Signal" << signum << "to" << executable_name_ << pid_ << "was successfully sent";

        if (signum == SIGTERM) {
            SetTerminatingState();
        }
    }

    // PID lookup will not be update as the process may still be running.
    // In case the pid is reused any existing usage of the pid will be
    // removed.
}

void StartupConfiguration::SetTerminatingState()
{
    SetState(ProcessState::kTerminating);
}

void StartupConfiguration::SetRunningState()
{
    SetState(ProcessState::kRunning);
}

void StartupConfiguration::SetState(ProcessState state)
{
    if (state != state_) {
        GetLogger().LogInfo() << "Process" << executable_name_ << "with pid" << pid_ << "changed its state from"
                              << GetProcessStateString(state_) << "to" << GetProcessStateString(state);
    }
    state_ = state;
}

void StartupConfiguration::SetTerminationCause(TerminationCause termination)
{
    if (termination != termination_) {
        GetLogger().LogDebug() << "Process" << executable_name_ << "with ID" << pid_
                               << "changed its termination cause from" << GetTerminationCauseString(termination_)
                               << "to" << GetTerminationCauseString(termination);
    }
    termination_ = termination;
}

std::string StartupConfiguration::GetName() const
{
    return executable_name_;
}

std::uint32_t StartupConfiguration::GetNumberOfRestartAttempts() const
{
    return number_of_restart_attempts_;
}

pid_t StartupConfiguration::GetPid() const
{
    return pid_;
}

std::string StartupConfiguration::GetConfigShortName() const noexcept
{
    return config_short_name_;
}

std::string StartupConfiguration::GetExecutable() const
{
    return executable_path_;
}

std::uint32_t StartupConfiguration::GetExecutionError() const noexcept
{
    return execution_error_;
}

const std::string& StartupConfiguration::GetFunctionClusterAffiliation() const noexcept
{
    return function_cluster_affiliation_;
}

const StartupConfiguration::DependencyCollectionType& StartupConfiguration::GetDependencies() const
{
    return dependencies_;
}

const std::string& StartupConfiguration::GetPrefix() const
{
    return prefix_;
}

int StartupConfiguration::GetSchedulingPolicyFromString(std::string const& policy)
{
    if ((policy == "SCHED_RR") || (policy == "Round Robin") || (policy == "SCHEDULING-POLICY-ROUND-ROBIN")) {
        return SCHED_RR;
    } else if ((policy == "SCHED_FIFO") || (policy == "FIFO") || (policy == "SCHEDULING-POLICY-FIFO")) {
        return SCHED_FIFO;
    } else {
        return SCHED_OTHER;
    }
}

ProcessState StartupConfiguration::GetState() const
{
    return state_;
}

TerminationBehavior StartupConfiguration::GetTerminationBehavior() const
{
    return termination_behavior_;
}

TerminationCause StartupConfiguration::GetTerminationCause() const
{
    return termination_;
}

std::string StartupConfiguration::GetProcessStateString(ProcessState state)
{
    std::string str;
    switch (state) {
    case ProcessState::kIdle:
        str = "kIdle";
        break;
    case ProcessState::kStarting:
        str = "kStarting";
        break;
    case ProcessState::kRunning:
        str = "kRunning";
        break;
    case ProcessState::kTerminating:
        str = "kTerminating";
        break;
    case ProcessState::kTerminated:
        str = "kTerminated";
        break;
    default:
        GetLogger().LogError() << "Invalid value of enum ProcessState";
        str = "Invalid";
        break;
    }
    return str;
}

std::string StartupConfiguration::GetTerminationCauseString(TerminationCause termination)
{
    std::string str;
    switch (termination) {
    case TerminationCause::kNormalExit:
        str = "kNormalExit";
        break;
    case TerminationCause::kAbnormalExit:
        str = "kAbnormalExit";
        break;
    default:
        str = "kUnknown";
        break;
    }
    return str;
}

bool StartupConfiguration::IsEnterTimeoutOver() const
{
    const auto& enterTimeout = enter_exit_timeout_.GetEnterTimeout().count();
    const auto& enterDuration = GetTimeSinceEpochInNs() - enter_timestamp_;

    if (ProcessState::kStarting == state_ && enterDuration > enterTimeout) {

        GetLogger().LogDebug() << "Detected enter timeout for process" << executable_name_ + ","
                               << "timeout:" << enterDuration << "ns";
        return true;
    }

    return false;
}

bool StartupConfiguration::IsExitTimeoutOver() const
{
    const auto& exitTimeout = enter_exit_timeout_.GetExitTimeout().count();
    const auto& exitDuration = GetTimeSinceEpochInNs() - exit_timestamp_;

    if (exitDuration > exitTimeout) {

        GetLogger().LogDebug() << "Detected exit timeout for process" << executable_name_ + ","
                               << "timeout:" << exitDuration << "ns";
        return true;
    }

    return false;
}

bool StartupConfiguration::IsPlatformApp() const
{
    return is_platform_app_;
}
#if 0
void StartupConfiguration::SendMessageToExecutionClient(const netaos::nosal::exec::ExecutionStateResponseStatus& responseStatus)
{
    std::string packedResponse;

    PackExecutionStateReportResponse(responseStatus, packedResponse);
    PipeMessage responseMessage{netaos::nosal::exec::internal::ExecutionClientOperation::kReportExecutionStateReponse,
        static_cast<std::uint32_t>(packedResponse.size()),
        {}};
    responseMessage.data.reserve(packedResponse.size());
    responseMessage.data.insert(responseMessage.data.cbegin(), packedResponse.cbegin(), packedResponse.cend());
    execStatePipeInput_.Write(responseMessage.header.tag, responseMessage.header.length, responseMessage.data.data());
}

void StartupConfiguration::ProcessExecStateReports()
{
    std::unique_ptr<PipeMessage> message = execStatePipeOutput_.Read();
    if (nullptr == message) {
        return;
    }

    GetLogger().LogDebug() << "Got a message from" << executable_name_ << pid_;

    if (message->header.tag != ::netaos::nosal::exec::internal::ExecutionClientOperation::kReportExecutionState
        || message->header.length != sizeof(ExecutionState)) {

        GetLogger().LogError() << "Invalid Execution State message";
        return;
    }

    netaos::nosal::exec::ExecutionStateResponseStatus responseStatus;

    const std::uint8_t executionState = message->data.front();

    if (reporting_behavior_ == config::ExecutionManifest::ExecutionStateReportingBehavior::kDoesNotReport) {
        responseStatus = netaos::nosal::exec::ExecutionStateResponseStatus::kReportNotAllowed;
        SendMessageToExecutionClient(responseStatus);
        return;
    }

    /// @uptrace{SWS_EM_01004, 4cd0b4994b483cf3ef85f64b445267fd3159de56}
    if (executionState == static_cast<std::uint8_t>(ExecutionState::kRunning)) {
        if (state_ == ProcessState::kStarting) {
            SetRunningState();
            responseStatus = netaos::nosal::exec::ExecutionStateResponseStatus::kSuccess;
        } else {
            GetLogger().LogError() << "Invalid ExecutionState Running report as process" << GetName()
                                   << "is not in Starting Process state.";
            /// @uptrace{SWS_EM_02243, 0bfa74b9f4952e113040a7402816e7d71715dd79}
            responseStatus = netaos::nosal::exec::ExecutionStateResponseStatus::kInvalidTransition;
        }
    } else {
        GetLogger().LogError() << "Broken Execution State message:" << executionState;
        responseStatus = netaos::nosal::exec::ExecutionStateResponseStatus::kInvalidReport;
    }

    SendMessageToExecutionClient(responseStatus);

    return;
}

void StartupConfiguration::ProcessExecutionStateReports(
    std::set<std::shared_ptr<StartupConfiguration>>& executedStartupConfigurations)
{
    // Poll executed processes for Execution State reports

    std::vector<std::shared_ptr<StartupConfiguration>> startup_configurations(executedStartupConfigurations.size());
    std::vector<pollfd> pollFds(startup_configurations.size());

    size_t idx = 0;
    for (std::shared_ptr<StartupConfiguration> startup_configuration : executedStartupConfigurations) {

        startup_configurations[idx] = startup_configuration;

        pollFds[idx].fd = startup_configuration->GetExecStatePipe().GetReadEndFd();
        pollFds[idx].events = POLLIN;
        pollFds[idx].revents = 0;

        ++idx;
    }

    sigset_t sigset;
    ::sigemptyset(&sigset);
    ::sigaddset(&sigset, SIGCHLD);

    int ret = ::ppoll(pollFds.data(), pollFds.size(), &kPollTimeout, &sigset);
    if (-1 == ret) {
        // If SIGCHLD interrupted the ppoll()...
        if (errno == EINTR) {
            // CheckProcesses should be called in anycase because of races (see below)
        } else {
            GetLogger().LogError() << "Failed to poll for ExecutionState reports, errno:" << errno;
        }
    } else if (ret > 0) {
        for (idx = 0; idx < pollFds.size(); ++idx) {
            if (pollFds[idx].revents & POLLIN) {
                pollFds[idx].revents = 0;

                startup_configurations[idx]->ProcessExecStateReports();
            }
        }
    } else {
        // timeout
    }

    // Poll Operating System for terminated child processes
    CheckProcesses(executedStartupConfigurations);
}

std::vector<std::pair<PipeMessage, std::shared_ptr<StartupConfiguration>>>
StartupConfiguration::PollForStateClientRequests(
    std::set<std::shared_ptr<StartupConfiguration>>& executedStartupConfigurations)
{
    std::vector<std::shared_ptr<StartupConfiguration>> startup_configurations(executedStartupConfigurations.size());
    std::vector<pollfd> pollFds(startup_configurations.size());

    size_t idx = 0;
    for (std::shared_ptr<StartupConfiguration> startup_configuration : executedStartupConfigurations) {

        startup_configurations[idx] = startup_configuration;

        pollFds[idx].fd = startup_configuration->GetStateClientRequestPipe().GetReadEndFd();
        pollFds[idx].events = POLLIN;
        pollFds[idx].revents = 0;

        ++idx;
    }

    std::vector<std::pair<PipeMessage, std::shared_ptr<StartupConfiguration>>> results;

    sigset_t sigset;
    ::sigemptyset(&sigset);
    ::sigaddset(&sigset, SIGCHLD);

    int ret = ::ppoll(pollFds.data(), pollFds.size(), &kPollTimeout, &sigset);
    if (-1 == ret) {
        // If SIGCHLD interrupted the ppoll()...
        if (errno == EINTR) {
            CheckProcesses(executedStartupConfigurations);
        } else {
            GetLogger().LogError() << "Failed to poll for State Client requests";
        }
    } else if (ret > 0) {
        for (idx = 0; idx < pollFds.size(); ++idx) {
            if (pollFds[idx].revents & POLLIN) {
                pollFds[idx].revents = 0;

                std::unique_ptr<PipeMessage> message = startup_configurations[idx]->stateClientRequestPipeOutput_.Read();
                if (message != nullptr) {
                    GetLogger().LogDebug() << "Got a StateClient message with tag" << message.get()->header.tag
                                           << "and length" << message.get()->header.length;

                    results.emplace_back(std::move(*(message.get())), startup_configurations[idx]);
                } else {
                    GetLogger().LogWarn() << "Got a non-complete StateClient message";
                }
            }
        }
    } else {
        // timeout / no requests
    }

    return results;
}

std::size_t StartupConfiguration::CheckProcesses(
    std::set<std::shared_ptr<StartupConfiguration>>& executedStartupConfigurations)
{
    GetLogger().LogDebug() << "Checking if there are terminated child processes";

    size_t terminatedCount = 0;

    // The loop will break if there is no terminated child
    while (true) {

        int wstatus;
        int waitpidResult = ::waitpid(-1, &wstatus, WNOHANG);

        if (-1 == waitpidResult) {
            if (ECHILD != errno) {
                GetLogger().LogError() << "waitpid() failed with errno " << errno;
            }

            break;
        } else if (waitpidResult == 0) {
            break;
        }

        // If some child process terminated then update its status
        ++terminatedCount;

        const int childPid = waitpidResult;
        auto hasPidPredicate
            = [childPid](const std::shared_ptr<StartupConfiguration> process) { return process->GetPid() == childPid; };

        const auto it = std::find_if(
            executedStartupConfigurations.cbegin(), executedStartupConfigurations.cend(), hasPidPredicate);
        if (it == executedStartupConfigurations.cend()) {
            GetLogger().LogError() << "Internal error: failed to find child" << childPid;
            break;
        }

        std::shared_ptr<StartupConfiguration> startup_configuration = *it;
        const ProcessState oldState = startup_configuration->state_;

        /// @uptrace{SWS_EM_01006, e3ce6e5c004628426145abacf2130269db54f94d}
        startup_configuration->SetState(ProcessState::kTerminated);

        const int oldPid = startup_configuration->pid_;
        startup_configuration->pid_ = 0;
        startup_configuration->exit_timestamp_ = GetTimeSinceEpochInNs();

        // Check why the process terminated
        if (WIFEXITED(wstatus)) {
            if ((ProcessState::kTerminating == oldState)
                || (ProcessState::kRunning == oldState
                    && TerminationBehavior::kProcessIsSelfTerminating == startup_configuration->termination_behavior_)) {
                startup_configuration->SetTerminationCause(TerminationCause::kNormalExit);
                GetLogger().LogInfo() << "Process" << startup_configuration->executable_name_
                                      << "was terminated normally";
            } else {
                /// @uptrace{SWS_EM_02314, 3d2e4a761a75688208fd67e2bb44183b4eff1b8c}
                startup_configuration->SetTerminationCause(TerminationCause::kAbnormalExit);
                GetLogger().LogWarn() << "Process" << startup_configuration->executable_name_
                                      << "terminated unexpectedly";
            }
        } else {
            startup_configuration->SetTerminationCause(TerminationCause::kAbnormalExit);
            const std::string errorDescription = GetUnexpectedTerminationCause(wstatus, oldState);
            GetLogger().LogError() << "Process" << startup_configuration->executable_name_ << errorDescription;
        }

        startup_configuration->ResetFindProcessLookup(oldPid);
    }

    return terminatedCount;
}

void StartupConfiguration::SendMessageToStateClient(PipeMessage&& message)
{
    GetLogger().LogInfo() << "Sending a message to State Client" << executable_name_.c_str() << pid_;
    stateClientResponsePipeInput_.Write(message.header.tag, message.header.length, message.data.data());
}

void StartupConfiguration::SetProcessSpecifier(netaos::nosal::core::Optional<ProcessSpecifier> process_specifier)
{
    process_specifier_ = process_specifier;
}

const netaos::nosal::core::Optional<ProcessSpecifier>& StartupConfiguration::GetProcessSpecifierRef() const
{
    return process_specifier_;
}

netaos::nosal::core::Optional<ProcessSpecifier> StartupConfiguration::GetProcessSpecifierFromManifest() const
{
    auto processNameRes = manifest_.GetProcessName();
    if (!processNameRes) {
        return netaos::nosal::core::nullopt;
    }
    auto res = ProcessSpecifier::Create(*processNameRes);
    if (!res) {
        return netaos::nosal::core::nullopt;
    }
    return res.Value();
}

void StartupConfiguration::UpdateFindProcessLookup(const pid_t pid)
{
    // Update the lookup
    if (process_specifier_.has_value()) {
        if (!netaos::nosal::exec::internal::findprocess::EmFindProcessWrapper::UpdateProcessPIDLookup(*process_specifier_, pid)) {
            GetLogger().LogFatal()
                << "StartupConfiguration::UpdateProcessPIDLookup: Unable to update find process lookup "
                   "with new pid. Terminating";
            std::abort();
        }
    } else {
        GetLogger().LogInfo() << "StartupConfiguration::UpdateProcessPIDLookup: \"" << executable_name_.c_str()
                              << "\" will not be updated as it cannot be identified";
    }
}

void StartupConfiguration::ResetFindProcessLookup(const pid_t pid)
{
    netaos::nosal::exec::internal::findprocess::EmFindProcessWrapper::ResetProcessPIDLookup(pid);
}

std::string StartupConfiguration::GetUnexpectedTerminationCause(int waitpidStatus,
    ProcessState stateBeforeTermination)
{
    std::string errorDescription;

    if (WIFSIGNALED(waitpidStatus)) {
        const int signal = WTERMSIG(waitpidStatus);
        switch (signal) {
        case SIGTERM:
            errorDescription = "failed to handle SIGTERM";

            if (ProcessState::kTerminating != stateBeforeTermination) {
                errorDescription += " sent by unknown process";
            }
            break;

        case SIGKILL:
            errorDescription = "was terminated by SIGKILL";
            break;

        case SIGABRT:
            errorDescription = "was terminated by SIGABRT";
            break;

        default:
            errorDescription = ("was terminated by a signal " + std::to_string(signal));
            break;
        }
    } else if (WIFSTOPPED(waitpidStatus)) {
        errorDescription
            = "was stopped by delivery of a signal that currently is not processed by EM and therefore the Process "
              "marked as terminated";
    } else if (WIFCONTINUED(waitpidStatus)) {
        errorDescription
            = "was resumed by delivery of SIGCONT that currently is not processed by EM and therefore the Process "
              "marked as terminated";
    } else {
        errorDescription = "was terminated abnormally";
    }

    return errorDescription;
}
#endif

TerminationBehavior StartupConfiguration::termationBehaviorFromString(
    const std::string& termination_behaviorString)
{
    if (termination_behaviorString == "PROCESS-IS-SELF-TERMINATING") {
        return TerminationBehavior::kProcessIsSelfTerminating;
    } else if (termination_behaviorString == "PROCESS-IS-NOT-SELF-TERMINATING") {
        return TerminationBehavior::kProcessIsNotSelfTerminating;
    } else {
        /// @uptrace{SWS_EM_01314, adf0690a3fdbeb0feac23614c36683c85d7f50eb}
        GetLogger().LogInfo()
            << "Process" << GetName()
            << "has no or wrong TerminationBehavior defined and will be treated as notSelfTerminating.";
        return TerminationBehavior::kProcessIsNotSelfTerminating;
    }
}

}  // namespace internal
}  // namespace exec
}  // namespace nosal
}  // namespace netaos
